package com.cnebula.dataprocess.sqoop.job.adapter;

import java.util.List;
import java.util.UUID;

import org.apache.sqoop.client.SqoopClient;
import org.apache.sqoop.model.MLink;
import org.apache.sqoop.model.MLinkConfig;
import org.apache.sqoop.validation.Message;
import org.apache.sqoop.validation.Status;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.cnebula.dataprocess.sqoop.job.exception.SqoopMLinkCreateException;
import com.cnebula.dataprocess.sqoop.job.model.LinkConfig;
import com.cnebula.dataprocess.sqoop.job.model.LspMLink;

public class MLinkJDBCAdapter extends SqoopMLinkAdapter{
	
	private static MLinkJDBCAdapter instance;
	
	private static final Logger log = LoggerFactory.getLogger(MLinkJDBCAdapter.class);
	
	static {
		instance = new MLinkJDBCAdapter();
	}
	
	private MLinkJDBCAdapter(){}
	
	public static MLinkJDBCAdapter getInstance(){
		return instance;
	}
	
	public LspMLink createMLink(SqoopClient client, LinkConfig config) throws SqoopMLinkCreateException{
		String linkName = "jdbc-link-" + UUID.randomUUID().toString().substring(0, 10);
		MLink mlink = client.createLink("generic-jdbc-connector");
		LspMLink link = new LspMLink(mlink);
        link.setName(linkName);
        link.setCreationUser("sqoop");
        link.setType(config.getType());
        link.setUrl(config.getUrl());
        
        MLinkConfig linkConfig = link.getConnectorLinkConfig();
        linkConfig.getStringInput("linkConfig.connectionString").setValue(config.getUrl());
        linkConfig.getStringInput("linkConfig.jdbcDriver").setValue(config.getDriver());
        linkConfig.getStringInput("linkConfig.username").setValue(config.getUserName());
        linkConfig.getStringInput("linkConfig.password").setValue(config.getPassword());
        linkConfig.getStringInput("dialect.identifierEnclose").setValue("`");
        Status status = client.saveLink(link);
        if (status.canProceed()) {
        	log.debug("Created Link with Link Name : " + link.getName());
            return link;
        } else {
        	List<Message> validationMessages = link.getValidationMessages();
        	String detailMsg = "";
        	for(Message msg : validationMessages){
        		if(!msg.getStatus().canProceed()){
        			detailMsg = msg.getMessage();
        			break;
        		}
        	}
        	throw new SqoopMLinkCreateException(detailMsg);
        }
	}
}
